package com.heima.kafkastatrm;

import com.alibaba.fastjson.JSON;
import com.heima.kafkastatrm.pojo.ArticleBehaviorMsg;
import com.heima.kafkastatrm.pojo.HotArticleConstants;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class BehaviorMsgProducer {

    private static final String KAFKA_IP = "192.168.137.136:9092";

    public static void main(String[] args) throws Exception{
        //添加kafka的配置信息
        Properties properties = new Properties();
        //配置broker信息
        properties.put("bootstrap.servers",KAFKA_IP);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.RETRIES_CONFIG,10);

        //生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);

        for(int i=0; i<10;i++){
            ArticleBehaviorMsg viewMsg = new ArticleBehaviorMsg();
            viewMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS);
            viewMsg.setArticleId(1663815807991365634L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(viewMsg)));

            ArticleBehaviorMsg likeMsg = new ArticleBehaviorMsg();
            likeMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.LIKES);
            likeMsg.setArticleId(1663815807991365634L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(likeMsg)));

            ArticleBehaviorMsg collectMsg = new ArticleBehaviorMsg();
            collectMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.COLLECTION);
            collectMsg.setArticleId(1663815807991365634L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(collectMsg)));

            ArticleBehaviorMsg commentMsg = new ArticleBehaviorMsg();
            commentMsg.setType(ArticleBehaviorMsg.ArticleBehaviorType.COMMENT);
            commentMsg.setArticleId(1663815807991365634L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(commentMsg)));

            ArticleBehaviorMsg viewMsg2 = new ArticleBehaviorMsg();
            viewMsg2.setType(ArticleBehaviorMsg.ArticleBehaviorType.VIEWS);
            viewMsg2.setArticleId(1664924339742453762L);
            producer.send(new ProducerRecord<String, String>(HotArticleConstants.HOT_ARTICLE_SCORE_STREAM_TOPIC, JSON.toJSONString(viewMsg2)));
            System.out.println("第"+(i+1)+"条消息发送完成");
            Thread.sleep(2000);
        }
        System.out.println("所有消息发送完成");
        //关系消息通道
        producer.close();
    }
}